package com.deng.eventflow.core.functions;

import com.deng.eventflow.core.EventTracker;
import com.deng.eventflow.core.Processor;
import com.deng.eventflow.core.PubliclyStateful;
import com.deng.eventflow.core.UniformProcessor;
import com.deng.eventflow.core.exception.FunctionException;
import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.petitpoucet.NodeFunction;
import com.deng.eventflow.core.petitpoucet.ProvenanceNode;

import java.util.Set;

public class ApplyFunction extends UniformProcessor implements PubliclyStateful {

    protected Function m_function;

    protected ShiftTracker m_shiftTracker;

    public ApplyFunction(Function comp) {
        super(comp.getInputArity(), comp.getOutputArity());
        m_function = comp;
        m_shiftTracker = new ShiftTracker();
    }

    @Override
    public void reset() {
        super.reset();
        m_function.reset();
    }

    @Override
    public synchronized ApplyFunction duplicate(boolean with_state) {
        ApplyFunction out = new ApplyFunction(m_function.duplicate(with_state));
        duplicateInto(out);
        return out;

    }

    @Override
    public final void getInputTypesFor( Set<Class<?>> classes, int index) {
        // The type is determined by that of the underlying function
        m_function.getInputTypesFor(classes, index);
    }

    @Override
    public final synchronized Class<?> getOutputType(int index) {
        // The type is determined by that of the underlying function
        return m_function.getOutputTypeFor(index);
    }

    @Override
    public String toString() {
        return m_function.toString();
    }

    public Function getFunction() {
        return m_function;
    }


    @Override
    public Object getState() {
        return 0;
    }

    public void cloneInto(ApplyFunction af, boolean with_state) {
        super.duplicateInto(af);
        af.m_function = m_function.duplicate(with_state);
    }

    @Override
    public Object printState() {
        return m_function;
    }


    @Override
    protected boolean compute(Object[] inputs, Object[] outputs) {
        try
        {
            m_function.evaluate(inputs, outputs, m_context, m_shiftTracker);
            m_inputCount++;
            m_outputCount++;
        }
        catch (FunctionException e)
        {
            throw new ProcessorException(e);
        }
        return true;
    }

    @Override
    public ApplyFunction readState(Object o) {
        Function f = (Function) o;
        return new ApplyFunction(f);
    }

    protected class ShiftTracker implements EventTracker {

        @Override
        public void associateTo(int id, NodeFunction f, int out_stream_index, int out_stream_pos)
        {
            if (m_eventTracker != null)
            {
                m_eventTracker.associateTo(getId(), f, out_stream_index, m_outputCount);
            }
        }

        @Override
        public void associateToInput(int id, int in_stream_index, int in_stream_pos,
                                     int out_stream_index, int out_stream_pos)
        {
            if (m_eventTracker != null)
            {
                m_eventTracker.associateToInput(getId(), in_stream_index, m_inputCount,
                        out_stream_index, m_outputCount);
            }
        }

        @Override
        public void associateToOutput(int id, int in_stream_index, int in_stream_pos,
                                      int out_stream_index, int out_stream_pos)
        {
            // Nothing to do
        }

        @Override
        public ProvenanceNode getProvenanceTree(int proc_id, int stream_index, int stream_pos)
        {
            throw new Error("ShiftTracker.getProvenanceTree should not be called");
        }

        @Override
        public void setConnection(int output_proc_id, int output_stream_index, int input_proc_id,
                                  int input_stream_index)
        {
            // Do nothing
        }

        @Override
        public void setTo(Processor ... processors)
        {
            // Do nothing
        }

        @Override
        public EventTracker getCopy()
        {
            return new ShiftTracker();
        }
    }
}
